package com.zhentao.service.impl;

import com.alibaba.fastjson.JSON;
import com.baomidou.mybatisplus.core.conditions.query.QueryWrapper;
import com.baomidou.mybatisplus.core.toolkit.StringUtils;
import com.baomidou.mybatisplus.extension.plugins.pagination.Page;
import com.baomidou.mybatisplus.extension.service.impl.ServiceImpl;
import com.zhentao.domain.Product;
import com.zhentao.domain.dto.ProductFindAllDto;
import com.zhentao.service.AuditService;
import com.zhentao.service.ProductService;
import com.zhentao.mapper.ProductMapper;
import com.zhentao.until.AppHttpCodeEnum;
import com.zhentao.until.ResponseResult;
import lombok.SneakyThrows;
import org.elasticsearch.action.bulk.BulkRequest;
import org.elasticsearch.action.index.IndexRequest;
import org.elasticsearch.action.search.SearchRequest;
import org.elasticsearch.action.search.SearchResponse;
import org.elasticsearch.client.RequestOptions;
import org.elasticsearch.client.RestHighLevelClient;
import org.elasticsearch.common.xcontent.XContentType;
import org.elasticsearch.index.query.BoolQueryBuilder;
import org.elasticsearch.index.query.Operator;
import org.elasticsearch.index.query.QueryBuilders;
import org.elasticsearch.index.query.QueryStringQueryBuilder;
import org.elasticsearch.search.SearchHit;
import org.elasticsearch.search.builder.SearchSourceBuilder;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.data.redis.core.RedisTemplate;
import org.springframework.kafka.core.KafkaTemplate;
import org.springframework.scheduling.annotation.Scheduled;
import org.springframework.stereotype.Service;

import java.util.ArrayList;
import java.util.Calendar;
import java.util.List;
import java.util.Set;

/**
* @author 薄超轩
* @description 针对表【product】的数据库操作Service实现
* @createDate 2024-09-07 21:38:34
*/
@Service
public class ProductServiceImpl extends ServiceImpl<ProductMapper, Product>
    implements ProductService{
    @Autowired
    private RestHighLevelClient restHighLevelClient;
    @Autowired
    private AuditService auditService;
    @Autowired
    private KafkaTemplate<String,Object> kafkaTemplate;
    @Autowired
    private RedisTemplate<String,Object> redisTemplate;

    @SneakyThrows
    @Override
    public ResponseResult findALl(ProductFindAllDto dto) {
        if (dto==null){
            return ResponseResult.errorResult(AppHttpCodeEnum.PARAM_INVALID);
        }
        if (StringUtils.isNotBlank(dto.getName())){
            SearchRequest searchRequest=new SearchRequest("product");
            SearchSourceBuilder searchSourceBuilder=new SearchSourceBuilder();
            BoolQueryBuilder builder= QueryBuilders.boolQuery();
            QueryStringQueryBuilder queryBuilder = QueryBuilders.queryStringQuery(dto.getName()).field("description")
                    .defaultOperator(Operator.OR);
            builder.must(queryBuilder);
            searchSourceBuilder.query(builder);
            searchRequest.source(searchSourceBuilder);
            SearchResponse search = restHighLevelClient.search(searchRequest, RequestOptions.DEFAULT);
            List<Product>list=new ArrayList<>();
            SearchHit[] hits = search.getHits().getHits();
            for (SearchHit hit : hits) {
                String sourceAsString = hit.getSourceAsString();
                Product product = JSON.parseObject(sourceAsString, Product.class);
                list.add(product);
            }
            return ResponseResult.okResult(list);
        }
        Page<Product>page=new Page<>(dto.getPage(),dto.getSize());
        QueryWrapper<Product>wrapper=new QueryWrapper<>();
        Page<Product> page1 = this.page(page, wrapper);
        return ResponseResult.okResult(page1.getRecords());
    }

    @Override
    public ResponseResult ProductSave(Product product) {
        if (product==null){
            return ResponseResult.errorResult(AppHttpCodeEnum.PARAM_INVALID);
        }
        this.save(product);
        String auditResult = auditService.auditProduct(product);
        if (auditResult.equals("审核通过")){
            //要上架的时间
            long auditTime = product.getAuditTime().getTime();

            //五分钟后的时间
            Calendar calendar = Calendar.getInstance();
            calendar.add(Calendar.MINUTE,5);
            long futureTime = calendar.getTimeInMillis();

            //当前时间
            long nowTime = System.currentTimeMillis();

            //判断如果大于当前时间就放到kafka里面去消费，否则放到zset里面
            if (auditTime<=nowTime){
                kafkaTemplate.send("productTopic", JSON.toJSONString(product));
                System.out.println("数据进入kafka"+product);
            }else if (auditTime<=futureTime){
                redisTemplate.opsForZSet().add("product:zset", product, product.getAuditTime().getTime());
                System.out.println("数据进入zset里面");
            }
        }
        return ResponseResult.okResult("成功");
    }



    @Scheduled(cron = "0 0/1 * * * ?")
    //开启一个定时任务将redis里面的数据同步到kafka里面
    public void zSetToKafka(){
        //获取一分钟后的时间
        Calendar calendar = Calendar.getInstance();
        calendar.add(Calendar.MINUTE,1);
        long futureTime = calendar.getTimeInMillis();

        //将一分钟后的数据从zSet里面拿出来
        Set<Object> set = redisTemplate.opsForZSet().rangeByScore("product:zset", 0, futureTime);
        for (Object o : set) {
            //将数据添加到kafka里面
            kafkaTemplate.send("productTopic",JSON.toJSONString(o));
            System.out.println("数据从zset同步到kafka"+o);
            redisTemplate.opsForZSet().remove("product:zset", o);
        }
    }
}




